package com.zhitan.ems.app.stat.service.impl;

import java.text.ParseException;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.QueryApi;
import com.influxdb.query.FluxRecord;
import com.influxdb.query.FluxTable;
import com.zhitan.ems.app.stat.vo.DeviceDataDateVo;
import com.zhitan.ems.common.extra.utils.DateUtil;
import com.zhitan.ems.common.utils.DateUtils;
import com.zhitan.ems.enums.StatType;

@Service
public class InfluxdbService {
  private static final Logger log = LoggerFactory.getLogger(InfluxdbService.class);
  
  @Autowired
  private InfluxDBClient influxdbClient;
  
  private static final DateTimeFormatter FORMATTER_SECONDS = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
  
  private static final DateTimeFormatter FORMATTER_MINUTES = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm");
  
  private static final DateTimeFormatter FORMATTER_HOURS = DateTimeFormatter.ofPattern("yyyy-MM-dd HH");
  
  private static final List<String> DEFAULT_IGNORE_KEYS = Arrays.asList(new String[] { "_stop", "_time", "_start", "_measurement", "result" });
  
  public static String formatInstant(Instant instant, int timeType) {
    if (instant == null)
      return ""; 
    ZonedDateTime zonedDateTime = instant.atZone(ZoneId.systemDefault());
    switch (timeType) {
      case 1:
        return zonedDateTime.format(FORMATTER_SECONDS);
      case 2:
        return zonedDateTime.format(FORMATTER_MINUTES);
      case 3:
        return zonedDateTime.format(FORMATTER_HOURS);
    } 
    return zonedDateTime.format(FORMATTER_SECONDS);
  }
  
  private static StringBuilder initializeQuery(String bucket) {
    StringBuilder query = new StringBuilder();
    query.append(String.format("from(bucket: \"%s\")", bucket));
    return query;
  }
  
  private static StringBuilder appendRange(StringBuilder query, String start, String end) {
    if (start != null && end != null) {
    	
      query.append(String.format(" |> range(start: %s, stop: %s)",  start, end));
    } else if (start != null) {
      query.append(String.format(" |> range(start: %s)", start));
    } else {
      query.append(" |> range(start: -1d)");
    } 
    return query;
  }
  
  private static StringBuilder appendFilterMeasurement(StringBuilder query, List<String> measurements) {
    if (measurements == null || measurements.isEmpty())
      return query; 
    if (measurements.size() == 1) {
      query.append(String.format(" |> filter(fn: (r) => r[\"_measurement\"] == \"%s\")", measurements.get(0)));
    } else {
      String conditions = measurements.stream().map(m -> String.format("r[\"_measurement\"] == \"%s\"",  m)).collect(Collectors.joining(" or "));
      query.append(String.format(" |> filter(fn: (r) => %s)", conditions));
    } 
    return query;
  }
  
  private static StringBuilder appendFilterMeasurement(StringBuilder query, String measurement) {
    if (measurement == null)
      return query; 
    query.append(String.format(" |> filter(fn: (r) => r[\"_measurement\"] == \"%s\")", measurement));
    return query;
  }
  
  private static StringBuilder appendFilterField(StringBuilder query, List<String> fields) {
    if (fields == null || fields.isEmpty())
      return query; 
    if (fields.size() == 1) {
      query.append(String.format(" |> filter(fn: (r) => r[\"_field\"] == \"%s\")", fields.get(0)));
    } else {
      String conditions = fields.stream().map(f -> String.format("r[\"_field\"] == \"%s\"",  f )).collect(Collectors.joining(" or "));
      query.append(String.format(" |> filter(fn: (r) => %s)", conditions));
    } 
    return query;
  }
  
  private static StringBuilder appendFilterTag(StringBuilder query, Map<String, String> tags) {
    if (tags == null || tags.isEmpty())
      return query; 
    for (Map.Entry<String, String> entry : tags.entrySet()) {
      query.append(String.format(" |> filter(fn: (r) => r[\"%s\"] == \"%s\")", entry.getKey(), entry.getValue()));
    } 
    return query;
  }
  
  private static StringBuilder appendStatType(StringBuilder query, StatType statType) {
    if (statType == null)
      return query; 
    return query.append(String.format(" |> %s()", new Object[] { statType.getCode() }));
  }
  
  public static List<Map<String, Object>> queryResultProcess(List<FluxTable> fluxTableList) {
    return queryResultProcess(fluxTableList, DEFAULT_IGNORE_KEYS);
  }
  
  public static List<Map<String, Object>> queryResultProcess(List<FluxTable> fluxTableList, List<String> ignoreKeyList) {
    if (fluxTableList == null || fluxTableList.isEmpty())
      return new ArrayList<>(); 
    List<Map<String, Object>> mapList = new ArrayList<>();
    for (FluxTable fluxTable : fluxTableList) {
      List<FluxRecord> fluxRecords = fluxTable.getRecords();
      if (fluxRecords != null && !fluxRecords.isEmpty())
        for (FluxRecord record : fluxRecords) {
          Map<String, Object> map = new HashMap<>(record.getValues());
          if (ignoreKeyList != null && !ignoreKeyList.isEmpty()) {
            Objects.requireNonNull(map);
            ignoreKeyList.forEach(map::remove);
          } 
          Instant timeInstant = record.getTime();
          map.put("dateStr", DateUtil.formatDateTime(DateUtil.toDateTime(timeInstant)));
          mapList.add(map);
        }  
    } 
    return mapList;
  }
  
  public static List<Map<String, Object>> queryResultProcess(List<FluxTable> fluxTableList, List<String> ignoreKeyList, Map<String, String> replaceKeyMap) {
    if (fluxTableList == null || fluxTableList.isEmpty())
      return new ArrayList<>(); 
    List<Map<String, Object>> mapList = new ArrayList<>();
    for (FluxTable fluxTable : fluxTableList) {
      List<FluxRecord> fluxRecords = fluxTable.getRecords();
      if (fluxRecords != null && !fluxRecords.isEmpty())
        for (FluxRecord record : fluxRecords) {
          Map<String, Object> map = new HashMap<>(record.getValues());
          if (ignoreKeyList != null && !ignoreKeyList.isEmpty()) {
            Objects.requireNonNull(map);
            ignoreKeyList.forEach(map::remove);
          } 
          if (replaceKeyMap != null && !replaceKeyMap.isEmpty())
            for (Map.Entry<String, String> entry : replaceKeyMap.entrySet()) {
              if (map.containsKey(entry.getKey())) {
                map.put(entry.getValue(), map.get(entry.getKey()));
                map.remove(entry.getKey());
              } 
            }  
          mapList.add(map);
        }  
    } 
    return mapList;
  }
  
  public static Map<String, Map<String, Object>> queryResultOneRowByTagProcess(List<FluxTable> fluxTableList, boolean devoiceCodeFag, int timeType) {
    if (fluxTableList == null || fluxTableList.isEmpty())
      return new HashMap<>(); 
    Map<String, Map<String, Object>> resultMap = new HashMap<>();
    for (FluxTable fluxTable : fluxTableList) {
      List<FluxRecord> fluxRecords = fluxTable.getRecords();
      if (fluxRecords != null && !fluxRecords.isEmpty())
        for (FluxRecord record : fluxRecords) {
          String timeKey = formatInstant(record.getTime(), timeType);
          Map<String, Object> itemData = resultMap.computeIfAbsent(timeKey, k -> new HashMap<>());
          if (record.getValues().containsKey("tag")) {
            String tagValue = (String)record.getValueByKey("tag");
            if (devoiceCodeFag && tagValue != null && tagValue.contains("_")) {
              int index = tagValue.indexOf("_");
              if (index != -1)
                tagValue = tagValue.substring(index + 1); 
            } 
            itemData.put(tagValue, record.getValue());
          } 
        }  
    } 
    return resultMap;
  }
  
  public static DeviceDataDateVo getResultOneRowByTagProcess(List<FluxTable> fluxTableList, boolean devoiceCodeFag) {
    if (fluxTableList == null || fluxTableList.isEmpty())
      return null; 
    DeviceDataDateVo dataDateVo = new DeviceDataDateVo();
    Instant time = null;
    Map<String, Object> resultMap = new HashMap<>();
    for (FluxTable fluxTable : fluxTableList) {
      List<FluxRecord> fluxRecords = fluxTable.getRecords();
      if (fluxRecords != null && !fluxRecords.isEmpty())
        for (FluxRecord record : fluxRecords) {
          if (time == null)
        	  time =record.getTime();
          if (record.getValues().containsKey("tag")) {
            String tagValue = (String)record.getValueByKey("tag");
            if (devoiceCodeFag && tagValue != null && tagValue.contains("_")) {
              int index = tagValue.indexOf("_");
              if (index != -1)
                tagValue = tagValue.substring(index + 1); 
            } 
            resultMap.put(tagValue, record.getValue());
            dataDateVo.setTag(tagValue);
            dataDateVo.setValue(record.getValue());
          } 
        }  
    } 
    dataDateVo.setData(resultMap);
    if (time != null) {
    	 ZonedDateTime zonedDateTime = time.atZone(ZoneId.systemDefault());
         dataDateVo.setDate(DateUtil.toDate(time));
         dataDateVo.setDateStr(zonedDateTime.format(FORMATTER_SECONDS));
    } 
    return dataDateVo;
  }
  public static List<DeviceDataDateVo> queryResultOneValueRowByTagProcess(List<FluxTable> fluxTableList, String tag, int timeType) {
    if (fluxTableList == null || fluxTableList.isEmpty())
      return new ArrayList<>(); 
    List<DeviceDataDateVo> list = new ArrayList<>();
    for (FluxTable fluxTable : fluxTableList) {
      List<FluxRecord> fluxRecords = fluxTable.getRecords();
      if (fluxRecords != null && !fluxRecords.isEmpty())
        for (FluxRecord record : fluxRecords) {
          String timeKey = formatInstant(record.getTime(), timeType);
          DeviceDataDateVo dataDateVo = new DeviceDataDateVo();
          dataDateVo.setDateStr(timeKey);
          if (tag != null) {
            dataDateVo.setTag(tag);
          } else if (record.getValues().containsKey("tag")) {
            dataDateVo.setTag(record.getValues().get("tag").toString());
          } 
          dataDateVo.setValue(record.getValue());
          list.add(dataDateVo);
        }  
    } 
    return list;
  }
  
  public static List<DeviceDataDateVo> queryResultOneRowToList(Map<String, Map<String, Object>> data, int timeType) {
    if (data == null || data.isEmpty())
      return new ArrayList<>(); 
    List<DeviceDataDateVo> list = new ArrayList<>();
    for (Map.Entry<String, Map<String, Object>> entry : data.entrySet()) {
      String key = entry.getKey();
      Map<String, Object> itemData = entry.getValue();
      DeviceDataDateVo dataDateVo = new DeviceDataDateVo();
      dataDateVo.setData(itemData);
      dataDateVo.setDateStr(key);
      if (key != null)
        try {
          switch (timeType) {
            case 1:
              dataDateVo.setDate(DateUtils.str_ToDate(key, "yyyy-MM-dd HH:mm:ss"));
              break;
            case 2:
              dataDateVo.setDate(DateUtils.str_ToDate(key, "yyyy-MM-dd HH:mm"));
              break;
            case 3:
              dataDateVo.setDate(DateUtils.str_ToDate(key, "yyyy-MM-dd HH"));
              dataDateVo.setDateStr(DateUtil.formatDateTime(dataDateVo.getDate()));
              break;
            default:
              dataDateVo.setDate(DateUtils.str_ToDate(key, "yyyy-MM-dd HH:mm:ss"));
              break;
          } 
        } catch (ParseException e) {
          log.warn("解析日期时出错:{}", key, e);
        }  
      list.add(dataDateVo);
    } 
    list.sort((d1, d2) -> d2.getDate().compareTo(d1.getDate()));
    return list;
  }
  
  public List<FluxTable> queryInfluxData(String bucket, String start, String end, String tableName, List<String> fields, HashMap<String, String> tags) {
    return queryInfluxData(bucket, start, end, Arrays.asList(tableName), fields, tags, null, null);
  }
  
  public List<FluxTable> queryInfluxData(String bucket, String start, String end, List<String> tableList, List<String> fields, HashMap<String, String> tags) {
    return queryInfluxData(bucket, start, end, tableList, fields, tags, null, null);
  }
  
  public List<FluxTable> queryInfluxData(String bucket, String start, String end, List<String> tableList, List<String> fields, HashMap<String, String> tags, Integer curPage, Integer pageSize) {
    try {
      StringBuilder queryBuilder = initializeQuery(bucket);
      queryBuilder = appendRange(queryBuilder, start, end);
      queryBuilder = appendFilterMeasurement(queryBuilder, tableList);
      if (fields != null && !fields.isEmpty())
        queryBuilder = appendFilterField(queryBuilder, fields); 
      if (tags != null && !tags.isEmpty())
        queryBuilder = appendFilterTag(queryBuilder, tags); 
      queryBuilder.append(" |> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")");
      queryBuilder.append(" |> group()");
      queryBuilder.append(" |> sort(columns: [\"_time\"], desc: true)");
      if (curPage != null && pageSize != null && curPage.intValue() > 0 && pageSize.intValue() > 0) {
        int offset = (curPage.intValue() - 1) * pageSize.intValue();
        queryBuilder.append(String.format(" |> limit(n: %d, offset: %d)", pageSize, Integer.valueOf(offset)));
      } 
      String queryString = queryBuilder.toString();
      log.info("-------queryInfluxData----query: {}", queryString);
      QueryApi queryApi = this.influxdbClient.getQueryApi();
      return queryApi.query(queryString);
    } catch (Exception e) {
      log.error("统计InfluxDB数据时出错:bucket={}, start={}, end={}, tables={}, fields={}, tags={}", new Object[] { bucket, start, end, tableList, fields, tags, e });
      throw new RuntimeException("统计InfluxDB数据失败", e);
    } 
  }
  
  public List<FluxTable> queryInfluxDataByDeviceCode(String bucket, String start, String end, List<String> tableList, String deviceCodeTag, Integer curPage, Integer pageSize,Integer statType) {
    return queryInfluxDataByDeviceCode(bucket, start, end, tableList, deviceCodeTag,null,curPage, pageSize, true,statType);
  }
  public List<FluxTable> queryInfluxDataByDeviceCode(String bucket, String start, String end, List<String> tableList, String deviceCodeTag,List<String> tags,  Integer curPage, Integer pageSize,Integer statType) {
	  return queryInfluxDataByDeviceCode(bucket, start, end, tableList, deviceCodeTag,tags,curPage, pageSize, true,statType);
  }
  
  public List<FluxTable> queryInfluxDataByDeviceCode(String bucket, String start, String end, List<String> tableList, String deviceCodeTag, List<String> tags,  Integer curPage, Integer pageSize, boolean isLike,Integer statType) {
    try {
      StringBuilder queryBuilder = initializeQuery(bucket);
      queryBuilder = appendRange(queryBuilder, start, end);
      queryBuilder = appendFilterMeasurement(queryBuilder, tableList);
      if(deviceCodeTag!=null) {
    	  if (isLike) {
    	        queryBuilder.append(String.format(" |> filter(fn: (r) => r.tag =~ /^%s/) ", deviceCodeTag));
    	      } else {
    	        queryBuilder.append(String.format(" |> filter(fn: (r) => r.tag == \"%s\")", deviceCodeTag));
    	      } 
      }
      if(tags!=null&&tags.size()>0) {
    	  String conditions = tags.stream().map(f -> String.format("r[\"tag\"] == \"%s\"",  f )).collect(Collectors.joining(" or "));
          queryBuilder.append(String.format(" |> filter(fn: (r) => %s)", conditions));
      }
      queryBuilder.append(" |> filter(fn: (r) => r[\"_field\"] == \"value\") ");
      
      if(statType!=null&&statType>0) {
    	  if(statType==1) {//小时
    		  queryBuilder.append( "|> aggregateWindow(every: 1h, fn: mean, createEmpty: false) ");
    	  }else if(statType==2) {//天
    		  queryBuilder.append( "|> aggregateWindow(every: 1d, fn: mean, createEmpty: false) ");
    	  }else if(statType==3) {//月 30 天
    		  queryBuilder.append( "|> aggregateWindow(every: 1mo, fn: mean, createEmpty: false) ");
    	  }else if(statType==4) {//年
    		  queryBuilder.append( "|> aggregateWindow(every: 1y, fn: mean, createEmpty: false) ");
    	  }
      }
      queryBuilder.append(" |> sort(columns: [\"_time\"], desc: true)");
      if (curPage != null && pageSize != null && curPage.intValue() > 0 && pageSize.intValue() > 0) {
        int offset = (curPage.intValue() - 1) * pageSize.intValue();
        queryBuilder.append(String.format(" |> limit(n: %d, offset: %d)", pageSize, Integer.valueOf(offset)));
      } 
      String queryString = queryBuilder.toString();
      log.info("-------queryInfluxDataByDeviceCode----query: {}", queryString);
      QueryApi queryApi = this.influxdbClient.getQueryApi();
      return queryApi.query(queryString);
    } catch (Exception e) {
    	log.error("统计InfluxDB数据时出错:bucket={}, start={}, end={}, tables={}, deviceCode={}", new Object[] { bucket, start, end, tableList, deviceCodeTag, e });
        throw new RuntimeException("统计InfluxDB数据失败", e);
    } 
  }
  
  public List<FluxTable> getFirstInfluxDataByDeviceCode(String bucket, String start, String end, List<String> tableList, String deviceCode) {
    try {
      StringBuilder queryBuilder = initializeQuery(bucket);
      queryBuilder = appendRange(queryBuilder, start, end);
      queryBuilder = appendFilterMeasurement(queryBuilder, tableList);
      queryBuilder.append(String.format(" |> filter(fn: (r) => r.tag =~ /^%s/) ", deviceCode));
      queryBuilder.append(" |> filter(fn: (r) => r[\"_field\"] == \"value\") ");
      queryBuilder.append(" |> first()");
      queryBuilder.append(" |> group()");
      String queryString = queryBuilder.toString();
      log.info("-------queryInfluxDataByDeviceCode----query: {}", queryString);
      QueryApi queryApi = this.influxdbClient.getQueryApi();
      return queryApi.query(queryString);
    } catch (Exception e) {
    	log.error("统计InfluxDB数据时出错:bucket={}, start={}, end={}, tables={}, deviceCode={}", new Object[] { bucket, start, end, tableList, deviceCode, e });
        throw new RuntimeException("统计InfluxDB数据失败", e);
    } 
  }
  
  public List<FluxTable> getLastInfluxDataByDeviceCode(String bucket, String start, String end, String table, String deviceCode) {
    return getLastInfluxDataByDeviceCode(bucket, start, end, Arrays.asList(table), deviceCode);
  }
  
  public List<FluxTable> getLastInfluxDataByDeviceCode(String bucket, String start, String end, List<String> tableList, String deviceCode) {
    try {
      StringBuilder queryBuilder = initializeQuery(bucket);
      queryBuilder = appendRange(queryBuilder, start, end);
      queryBuilder = appendFilterMeasurement(queryBuilder, tableList);
      queryBuilder.append(String.format(" |> filter(fn: (r) => r.tag =~ /^%s/) ", deviceCode));
      queryBuilder.append(" |> filter(fn: (r) => r[\"_field\"] == \"value\") ");
      queryBuilder.append(" |> last()");
      queryBuilder.append(" |> group()");
      String queryString = queryBuilder.toString();
      log.info("-------queryInfluxDataByDeviceCode----query: {}", queryString);
      QueryApi queryApi = this.influxdbClient.getQueryApi();
      return queryApi.query(queryString);
    } catch (Exception e) {
    	log.error("统计InfluxDB数据时出错:bucket={}, start={}, end={}, tables={}, deviceCode={}", new Object[] { bucket, start, end, tableList, deviceCode, e });
        throw new RuntimeException("统计InfluxDB数据失败", e);
    } 
  }
  public List<FluxTable> getLastInfluxDataByDeviceCodePoint(String bucket, String start, String end, String table, String deviceCodePointCode) {
	  return getLastInfluxDataByDeviceCodePoint(bucket, start, end, Arrays.asList(table), deviceCodePointCode);
  }
  
  public List<FluxTable> getLastInfluxDataByDeviceCodePoint(String bucket, String start, String end, List<String> tableList, String deviceCodePointCode) {
	  try {
		  StringBuilder queryBuilder = initializeQuery(bucket);
		  queryBuilder = appendRange(queryBuilder, start, end);
		  queryBuilder = appendFilterMeasurement(queryBuilder, tableList);
		  queryBuilder.append(String.format(" |> filter(fn: (r) => r.tag == \"%s\")", deviceCodePointCode));
		  
		  queryBuilder.append(" |> filter(fn: (r) => r[\"_field\"] == \"value\") ");
		  queryBuilder.append(" |> last()");
		  queryBuilder.append(" |> group()");
		  String queryString = queryBuilder.toString();
		  log.info("-------queryInfluxDataByDeviceCode----query: {}", queryString);
		  QueryApi queryApi = this.influxdbClient.getQueryApi();
		  return queryApi.query(queryString);
	  } catch (Exception e) {
		  log.error("统计InfluxDB数据时出错:bucket={}, start={}, end={}, tables={}, deviceCode={}", new Object[] { bucket, start, end, tableList, deviceCodePointCode, e });
		  throw new RuntimeException("统计InfluxDB数据失败", e);
	  } 
  }
  
  public long countInfluxDataByDeviceCode(String bucket, String start, String end, List<String> tableList, String deviceCode, boolean isGroup,Integer statType) {
    return countInfluxDataByDeviceCode(bucket, start, end, tableList, deviceCode, isGroup, true,statType);
  }
  public long countInfluxDataByDeviceCode(String bucket, String start, String end, List<String> tableList, String deviceCodeTag, boolean isGroup, boolean isLike,Integer statType) {
    try {
      StringBuilder queryBuilder = initializeQuery(bucket);
      queryBuilder = appendRange(queryBuilder, start, end);
      queryBuilder = appendFilterMeasurement(queryBuilder, tableList);
      if (isLike) {
        queryBuilder.append(String.format(" |> filter(fn: (r) => r.tag =~ /^%s/) ",deviceCodeTag));
      } else {
        queryBuilder.append(String.format(" |> filter(fn: (r) => r.tag == \"%s\")",deviceCodeTag));
      } 
      queryBuilder.append(" |> filter(fn: (r) => r[\"_field\"] == \"value\") ");
      if(statType!=null&&statType>0) {
    	  if(statType==1) {//小时
    		  queryBuilder.append( "|> aggregateWindow(every: 1h, fn: mean, createEmpty: false) ");
    	  }else if(statType==2) {//天
    		  queryBuilder.append( "|> aggregateWindow(every: 1d, fn: mean, createEmpty: false) ");
    	  }else if(statType==3) {//月 30 天
    		  queryBuilder.append( "|> aggregateWindow(every: 1mo, fn: mean, createEmpty: false) ");
    	  }else if(statType==4) {//年
    		  queryBuilder.append( "|> aggregateWindow(every: 1y, fn: mean, createEmpty: false) ");
    	  }
      }
      if (isGroup) queryBuilder.append(" |> group() "); 
      queryBuilder.append(" |> count() ");
      queryBuilder.append(" |> yield(name: \"total_count\") ");
      String queryString = queryBuilder.toString();
      log.info("-------countInfluxData----query: {}", queryString);
      QueryApi queryApi = this.influxdbClient.getQueryApi();
      List<FluxTable> fluxTableList = queryApi.query(queryString);
      if (fluxTableList != null && !fluxTableList.isEmpty()) {
        List<FluxRecord> fluxRecords = ((FluxTable)fluxTableList.get(0)).getRecords();
        if (fluxRecords != null && !fluxRecords.isEmpty())
          return Long.parseLong(((FluxRecord)fluxRecords.get(0)).getValue().toString()); 
      } 
      return 0L;
    } catch (Exception e) {
      log.error("统计InfluxDB数据时出错:bucket={}, start={}, end={}, tables={}, deviceCode={}", new Object[] { bucket, start, end, tableList, deviceCodeTag, e });
      throw new RuntimeException("统计InfluxDB数据失败", e);
    } 
  }
  
  private static StringBuilder buildBaseQuery(String bucket, String start, String end, List<String> tableList, List<String> fields, HashMap<String, String> tags, Boolean isGroup) {
    StringBuilder queryBuilder = initializeQuery(bucket);
    queryBuilder = appendRange(queryBuilder, start, end);
    queryBuilder = appendFilterMeasurement(queryBuilder, tableList);
    if (fields != null && !fields.isEmpty())
      queryBuilder = appendFilterField(queryBuilder, fields); 
    if (tags != null && !tags.isEmpty())
      queryBuilder = appendFilterTag(queryBuilder, tags); 
    if (isGroup.booleanValue())
      queryBuilder.append(" |> group()"); 
    return queryBuilder;
  }
  
  private List<FluxTable> executeAggregationQuery(String bucket, String start, String end, List<String> tableList, List<String> fields, HashMap<String, String> tags, String aggregation) {
    try {
      StringBuilder queryBuilder = buildBaseQuery(bucket, start, end, tableList, fields, tags, Boolean.valueOf(false));
      queryBuilder.append(String.format(" |> %s()", aggregation));
      String queryString = queryBuilder.toString();
      log.debug("{}{}", aggregation, queryString);
      QueryApi queryApi = this.influxdbClient.getQueryApi();
      return queryApi.query(queryString);
    } catch (Exception e) {
      log.error("执行{}聚合查询时出错:bucket={}, start={}, end={}, tables={}, fields={}, tags={}", new Object[] { aggregation, bucket, start, end, tableList, fields, tags, e });
      throw new RuntimeException("执行"+ aggregation +"聚合查询失败", e);
    } 
  }
  
  public List<FluxTable> querySum(String bucket, String start, String end, List<String> tableList, List<String> fields, HashMap<String, String> tags) {
    return executeAggregationQuery(bucket, start, end, tableList, fields, tags, "sum");
  }
  
  public List<FluxTable> queryMean(String bucket, String start, String end, List<String> tableList, List<String> fields, HashMap<String, String> tags) {
    return executeAggregationQuery(bucket, start, end, tableList, fields, tags, "mean");
  }
  
  public List<FluxTable> queryMax(String bucket, String start, String end, List<String> tableList, List<String> fields, HashMap<String, String> tags) {
    return executeAggregationQuery(bucket, start, end, tableList, fields, tags, "max");
  }
  
  public List<FluxTable> queryMin(String bucket, String start, String end, List<String> tableList, List<String> fields, HashMap<String, String> tags) {
    return executeAggregationQuery(bucket, start, end, tableList, fields, tags, "min");
  }
  
  public List<FluxTable> queryCount(String bucket, String start, String end, List<String> tableList, List<String> fields, HashMap<String, String> tags) {
    return executeAggregationQuery(bucket, start, end, tableList, fields, tags, "count");
  }
  
  public List<FluxTable> queryAggregations(String bucket, String start, String end, String table, List<String> fields, HashMap<String, String> tags) {
    return queryAggregations(bucket, start, end, Arrays.asList(table), fields, tags, null, null, null);
  }
  
  public List<FluxTable> queryAggregations(String bucket, String start, String end, String table, List<String> fields, HashMap<String, String> tags, Integer curPage, Integer pageSize) {
    return queryAggregations(bucket, start, end, Arrays.asList(table), fields, tags, null, curPage, pageSize);
  }
  
  public List<FluxTable> queryAggregations(String bucket, String start, String end, List<String> tableList, List<String> fields, HashMap<String, String> tags, List<String> aggregations, Integer curPage, Integer pageSize) {
    try {
      StringBuilder queryBuilder = initializeQuery(bucket);
      queryBuilder = appendRange(queryBuilder, start, end);
      queryBuilder = appendFilterMeasurement(queryBuilder, tableList);
      if (fields != null && !fields.isEmpty())
        queryBuilder = appendFilterField(queryBuilder, fields); 
      if (tags != null && !tags.isEmpty())
        queryBuilder = appendFilterTag(queryBuilder, tags); 
      queryBuilder.append(" |> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")");
      queryBuilder.append(" |> group()");
      if (aggregations != null && !aggregations.isEmpty())
        for (String aggregation : aggregations) {
          queryBuilder.append(String.format(" |> %s()", aggregation));
        }  
      if (curPage != null && pageSize != null && curPage.intValue() > 0 && pageSize.intValue() > 0) {
        int offset = (curPage.intValue() - 1) * pageSize.intValue();
        queryBuilder.append(String.format(" |> limit(n: %d, offset: %d)",  pageSize, Integer.valueOf(offset) ));
      } 
      String queryString = queryBuilder.toString();
      log.info("-------queryAggregations----query: {}", queryString);
      QueryApi queryApi = this.influxdbClient.getQueryApi();
      return queryApi.query(queryString);
    } catch (Exception e) {
      log.error("执行聚合查询时出错:bucket={}, start={}, end={}, tables={}, fields={}, tags={}, aggregations={}", new Object[] { bucket, start, end, tableList, fields, tags, aggregations, e });
      throw new RuntimeException("执行聚合查询失败", e);
    } 
  }
  
  public List<FluxTable> queryAggregationByWindow(String bucket, String start, String end, List<String> tableList, List<String> fields, HashMap<String, String> tags, String window, String aggregation) {
    try {
      StringBuilder queryBuilder = initializeQuery(bucket);
      queryBuilder = appendRange(queryBuilder, start, end);
      queryBuilder = appendFilterMeasurement(queryBuilder, tableList);
      if (fields != null && !fields.isEmpty())
        queryBuilder = appendFilterField(queryBuilder, fields); 
      if (tags != null && !tags.isEmpty())
        queryBuilder = appendFilterTag(queryBuilder, tags); 
      queryBuilder.append(" |> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")");
      queryBuilder.append(String.format(" |> window(every: %s)",window));
      queryBuilder.append(String.format(" |> %s()", aggregation));
      String queryString = queryBuilder.toString();
      log.info("-------queryAggregationByWindow----query: {}", queryString);
      QueryApi queryApi = this.influxdbClient.getQueryApi();
      return queryApi.query(queryString);
    } catch (Exception e) {
      log.error("按时间窗口执行聚合查询时出错:bucket={}, start={}, end={}, tables={}, fields={}, tags={}, window={}, aggregation={}", new Object[] { bucket, start, end, tableList, fields, tags, window, aggregation, e });
      throw new RuntimeException("按时间窗口执行聚合查询失败", e);
    } 
  }
  
  public List<FluxTable> queryFirst(String bucket, String start, String end, String table, List<String> fields, HashMap<String, String> tags, String valueField) {
    return queryFirst(bucket, start, end, Arrays.asList(table ), fields, tags);
  }
  
  public List<FluxTable> queryFirst(String bucket, String start, String end, List<String> tableList, List<String> fields, HashMap<String, String> tags) {
    try {
      StringBuilder queryBuilder = initializeQuery(bucket);
      queryBuilder = appendRange(queryBuilder, start, end);
      queryBuilder = appendFilterMeasurement(queryBuilder, tableList);
      if (fields != null && !fields.isEmpty())
        queryBuilder = appendFilterField(queryBuilder, fields); 
      if (tags != null && !tags.isEmpty())
        queryBuilder = appendFilterTag(queryBuilder, tags); 
      queryBuilder.append(" |> first()");
      queryBuilder.append(" |> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")");
      queryBuilder.append(" |> group()");
      String firstQuery = queryBuilder.toString();
      QueryApi queryApi = this.influxdbClient.getQueryApi();
      return queryApi.query(firstQuery);
    } catch (Exception e) {
      log.error("计算首尾数据差值时出错:bucket={}, start={}, end={}, tables={}, fields={}, tags={}, valueField={}", new Object[] { bucket, start, end, tableList, fields, tags, e });
      throw new RuntimeException("计算首尾数据差值失败", e);
    } 
  }
  
  public List<FluxTable> queryLast(String bucket, String start, String end, String table, List<String> fields, HashMap<String, String> tags) {
    return queryLast(bucket, start, end, Arrays.asList(new String[] { table } ), fields, tags);
  }
  
  public List<FluxTable> queryLast(String bucket, String start, String end, List<String> tableList, List<String> fields, HashMap<String, String> tags) {
    try {
      StringBuilder queryBuilder = initializeQuery(bucket);
      queryBuilder = appendRange(queryBuilder, start, end);
      queryBuilder = appendFilterMeasurement(queryBuilder, tableList);
      if (fields != null && !fields.isEmpty())
        queryBuilder = appendFilterField(queryBuilder, fields); 
      if (tags != null && !tags.isEmpty())
        queryBuilder = appendFilterTag(queryBuilder, tags); 
      queryBuilder.append(" |> last()");
      queryBuilder.append(" |> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")");
      queryBuilder.append(" |> group()");
      String lastQuery = queryBuilder.toString();
      log.info("====lastQuery========" + lastQuery);
      QueryApi queryApi = this.influxdbClient.getQueryApi();
      return queryApi.query(lastQuery);
    } catch (Exception e) {
      log.error("计算首尾数据差值是出错:bucket={}, start={}, end={}, tables={}, fields={}, tags={}", new Object[] { bucket, start, end, tableList, fields, tags,e});
      throw new RuntimeException("计算首尾数据差值失败", e);
    } 
  }
}
